package com.kiss.rpc.proxy.handler;

import com.kiss.rpc.NettyBootstrapInitializer;
import com.kiss.rpc.RpcBootstrap;
import com.kiss.rpc.annotation.TryTimes;
import com.kiss.rpc.compress.CompressorFactory;
import com.kiss.rpc.discovery.Registry;
import com.kiss.rpc.enumeration.RequestType;
import com.kiss.rpc.exceptions.DiscoveryException;
import com.kiss.rpc.exceptions.NetworkException;
import com.kiss.rpc.protection.CircuitBreaker;
import com.kiss.rpc.serialize.SerializerFactory;
import com.kiss.rpc.transport.message.RequestPayload;
import com.kiss.rpc.transport.message.RpcRequest;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import lombok.extern.slf4j.Slf4j;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.net.InetSocketAddress;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

@Slf4j
public class RpcConsumerInvocationHandler implements InvocationHandler {

    private final Class<?> interfaceRef;
    private final String group;

    public RpcConsumerInvocationHandler(Registry registry, Class<?> interfaceRef, String group) {
        this.interfaceRef = interfaceRef;
        this.group = group;
    }

    @Override
    public Object invoke(Object proxy, Method method, Object[] args) {
        TryTimes tryTimesAnnotation = method.getAnnotation(TryTimes.class);

        // 默认值0,代表不重试
        int tryTimes = 0;
        int intervalTime = 0;
        if (tryTimesAnnotation != null) {
            tryTimes = tryTimesAnnotation.tryTimes();
            intervalTime = tryTimesAnnotation.intervalTime();
        }

        int oriTimes = tryTimes;

        while (true) {

            // 1、封装报文
            RequestPayload requestPayload = RequestPayload.builder()
                    .interfaceName(interfaceRef.getName())
                    .methodName(method.getName())
                    .parametersType(method.getParameterTypes())
                    .parametersValue(args)
                    .returnType(method.getReturnType())
                    .build();

            // 创建请求
            RpcRequest rpcRequest = RpcRequest.builder()
                    .requestId(RpcBootstrap.getInstance().getConfig().getIdGenerator().getId())
                    .compressType(CompressorFactory.getCompressor(RpcBootstrap.getInstance().getConfig().getCompressType()).getCode())
                    .serializeType(SerializerFactory.getSerializer(RpcBootstrap.getInstance().getConfig().getSerializeType()).getCode())
                    .requestType(RequestType.REQUEST.getId())
                    .timeStamp(System.currentTimeMillis())
                    .requestPayload(requestPayload)
                    .build();

            RpcBootstrap.REQUEST_THREAD_LOCAL.set(rpcRequest);

            InetSocketAddress address = RpcBootstrap.getInstance()
                    .getConfig().getLoadBalancer().selectServiceAddress(interfaceRef.getName(), group);

            if (log.isDebugEnabled()) {
                log.debug("服务调用方，发现了服务【{}】的可用主机【{}】.",
                        interfaceRef.getName(), address);
            }

            CircuitBreaker circuitBreaker = RpcBootstrap.getInstance().getConfig().getEveryIpCircuitBreaker().get(address);
            if (circuitBreaker == null) {
                circuitBreaker = new CircuitBreaker(10, 0.5F);
                RpcBootstrap.getInstance().getConfig().getEveryIpCircuitBreaker().put(address, circuitBreaker);
            }

            try {
                if (rpcRequest.getRequestType() != RequestType.HEART_BEAT.getId() && circuitBreaker.isBreak()) {
                    Timer timer = new Timer();
                    timer.schedule(new TimerTask() {
                        @Override
                        public void run() {
                            RpcBootstrap.getInstance().getConfig().getEveryIpCircuitBreaker().get(address).reset();
                        }
                    }, 5000);

                    throw new RuntimeException("当前断路器已经开启，无法发送请求");
                }

                Channel channel = getAvailableChannel(address);

                if (log.isDebugEnabled()) {
                    log.debug("获取了和【{}】建立的连接通道,准备发送数据.", address);
                }

                CompletableFuture<Object> completableFuture = new CompletableFuture<>();
                RpcBootstrap.PENDING_REQUEST.put(rpcRequest.getRequestId(), completableFuture);

                channel.writeAndFlush(rpcRequest).addListener(
                        (ChannelFutureListener) future -> {
                            if (!future.isSuccess()) {
                                completableFuture.completeExceptionally(future.cause());
                            }
                        }
                );

                RpcBootstrap.REQUEST_THREAD_LOCAL.remove();

                Object res = completableFuture.get(10, TimeUnit.SECONDS);
                circuitBreaker.recordRequest();

                return res;

            } catch (Exception e) {

                tryTimes--;
                circuitBreaker.recordErrorRequest();
                try {
                    Thread.sleep(intervalTime);
                } catch (InterruptedException ex) {
                    log.error("在进行重试时发生异常.", ex);
                }
                if (tryTimes < 0) {
                    log.error("对方法【{}】进行远程调用时，重试{}次，依然不可调用",
                            method.getName(), oriTimes, e);
                    break;
                }
                log.error("在进行第{}次重试时发生异常.", 3 - tryTimes, e);
            }

        }
        throw new RuntimeException("执行远程方法" + method.getName() + "调用失败。");
    }

    private Channel getAvailableChannel(InetSocketAddress address) {

        Channel channel = RpcBootstrap.CHANNEL_CACHE.get(address);

        if (channel == null) {
            CompletableFuture<Channel> completableFuture = new CompletableFuture<>();
            NettyBootstrapInitializer.getBootstrap().connect(address).addListener(
                    (ChannelFutureListener) future -> {
                        if (future.isDone()) {
                            if (log.isDebugEnabled()) {
                                log.debug("客户端已经和【{}】建立了连接", address);
                            }
                            completableFuture.complete(future.channel());
                        } else if (!future.isSuccess()) {
                            completableFuture.completeExceptionally(future.cause());
                        }
                    }
            );
            try {
                channel = completableFuture.get();
            } catch (Exception e) {
                log.error("客户端获取与服务端的通道时，发生异常。", e);
                throw new DiscoveryException(e);
            }

            RpcBootstrap.CHANNEL_CACHE.put(address, channel);
        }

        if (channel == null) {
            log.error("获取或建立与【{}】的通道时发生了异常。", address);
            throw new NetworkException("获取通道时发生了异常。");
        }

        return channel;
    }
}
